/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package prictice.trident;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.LocalDRPC;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.trident.TridentState;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.builtin.Count;
import org.apache.storm.trident.operation.builtin.FilterNull;
import org.apache.storm.trident.operation.builtin.FirstN;
import org.apache.storm.trident.operation.builtin.MapGet;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.trident.testing.MemoryMapState;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import prictice.trident.functions.MySplit;
import prictice.trident.functions.Split;

import java.util.Random;


public class TridentPVTopo {

    public static StormTopology buildTopology(LocalDRPC drpc) {

        Random random = new Random();
        String[] host = {"www.taobao.com"};
        String[] sessionId = {"HJKHKJHUIFDISH21312321", "JKHIHNIDFSFDS34535", "FDSFDSFDSFDSMKMK908", "DFSFDSFDSAAAA1123245", "GFGDGFDGHKINNIU778"};
        String[] time = {"2014-01-07 08:40:50", "2014-01-07 08:40:50",
                "2014-02-07 06:40:50", "2014-01-27 06:40:50",
                "2014-03-07 09:40:50", "2014-01-17 11:40:50",
                "2014-04-07 08:40:50", "2014-11-08 12:40:50",
                "2014-05-07 08:40:50"};

        FixedBatchSpout spout = new FixedBatchSpout(new Fields("eachLog"), 3,
                new Values(host[0] + "\t" + sessionId[random.nextInt(5)] + "\t" + time[random.nextInt(9)]),
                new Values(host[0] + "\t" + sessionId[random.nextInt(5)] + "\t" + time[random.nextInt(9)]),
                new Values(host[0] + "\t" + sessionId[random.nextInt(5)] + "\t" + time[random.nextInt(9)]),
                new Values(host[0] + "\t" + sessionId[random.nextInt(5)] + "\t" + time[random.nextInt(9)]),
                new Values(host[0] + "\t" + sessionId[random.nextInt(5)] + "\t" + time[random.nextInt(9)]),
                new Values(host[0] + "\t" + sessionId[random.nextInt(5)] + "\t" + time[random.nextInt(9)]),
                new Values(host[0] + "\t" + sessionId[random.nextInt(5)] + "\t" + time[random.nextInt(9)]),
                new Values(host[0] + "\t" + sessionId[random.nextInt(5)] + "\t" + time[random.nextInt(9)]));
        spout.setCycle(true);

        TridentTopology topology = new TridentTopology();
        TridentState wordCounts = topology.newStream("spout1", spout)
                //.parallelismHint(16)
                .each(new Fields("eachLog"), new MySplit("\t"), new Fields("date", "sessionId"))
                .groupBy(new Fields("date"))
                .persistentAggregate(new MemoryMapState.Factory(), new Fields("sessionId"), new Count(), new Fields("PV"));
        //   .parallelismHint(16);

        topology.newDRPCStream("GetPv", drpc)
                .each(new Fields("args"), new Split(" "), new Fields("date"))
                .groupBy(new Fields("date"))
                .stateQuery(wordCounts, new Fields("date"), new MapGet(), new Fields("PV"))
                .each(new Fields("PV"), new FilterNull())
                .applyAssembly(new FirstN(2,"PV",true));
               // .project(new Fields("word", "count"));
        return topology.build();
    }

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
        conf.setMaxSpoutPending(20);
        conf.setDebug(false);
        if (args.length == 0) {
            LocalDRPC drpc = new LocalDRPC();
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("wordCounter", conf, buildTopology(drpc));
            for (int i = 0; i < 100; i++) {
                System.err.println("DRPC RESULT: " + drpc.execute("GetPv", "2014-01-27 2014-03-07"));
                Thread.sleep(1000);
            }
        } else {
            conf.setNumWorkers(3);
            StormSubmitter.submitTopologyWithProgressBar(args[0], conf, buildTopology(null));
        }
    }
}
